package com.basker.pisces.domain.command.run;

import java.util.List;
import java.util.Set;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.stereotype.Component;
import org.springframework.util.Assert;

import com.basker.pisces.core.exception.StaleObjectException;
import com.basker.pisces.core.meta.IDataField;
import com.basker.pisces.core.meta.IDataObjectMeta;
import com.basker.pisces.core.repository.IDataRepository;
import com.basker.pisces.domain.command.CommandExecuteContext;
import com.basker.pisces.domain.command.CommandResult;
import com.basker.pisces.domain.command.exception.CommandExecuteInvalidObjectException;
import com.basker.pisces.domain.command.exception.CommandExecuteRuntimeException;
import com.basker.pisces.domain.command.meta.Command;
import com.basker.pisces.domain.command.option.CommandExecuteOption;
import com.basker.pisces.domain.entity.meta.Entity;
import com.basker.pisces.domain.event.CommandExecutedDomainEvent;
import com.basker.pisces.domain.validation.ValidationErrors;
import com.basker.pisces.resource.StringResources;
import com.basker.pisces.tx.TransactionHelper;

/**
 * 命令执行器
 *
 * @author hangwen
 */
@Component
public final class CommandExecutor {

    private static final Logger logger = LoggerFactory.getLogger(CommandExecutor.class);
    private static final CommandExecuteOption DEFAULT_OPTION = CommandExecuteOption.DEFAULT_OPTION;

    @Autowired
    private IDataRepository repository;

    /**
     * spring事件发布器
     */
    @Autowired
    private ApplicationEventPublisher applicationEventPublisher;

    /**
     * 对实体执行命令，传入主键
     *
     * @param entity
     * @param command
     * @param pkValues
     * @return
     */
    public CommandResult execute(Entity entity, Command command, List<Object> pkValues) {
        return execute(entity, command, pkValues, DEFAULT_OPTION);
    }

    /**
     * 对实体执行命令，传入主键和命令执行选项
     *
     * @param entity
     * @param command
     * @param pkValues
     * @param option
     * @return
     */
    public CommandResult execute(Entity entity, Command command, List<Object> pkValues, CommandExecuteOption option) {
        assertParameters(entity, command);

        final CommandExecuteOption option2Use = option == null ? DEFAULT_OPTION : option;

        // 1.初始化命令运行器
        // 2.处理实体：对于有些实体，只需要部分属性
        CommandRunner<?> runner = CommandRunnerFactory.createRunner(entity, command, option2Use);

        // 3.如果命令需要加锁，需要再加载数据前先加锁，失败直接返回错误
        if (runner.getCommand().getEnableLock()) {
            CommandResult result = this.tryLock(runner, pkValues, option2Use);
            if (result != null) {
                return result;
            }
        }

        try {
            CommandResult result = this.executeByPkValues(runner, pkValues, option2Use);

            if (!result.isSuccess() && command.getEnableRetryWhileObjectStale()) {
                // 后续看需要是否需要动态设置重试次数
                int retryCount = 3;
                RetryExecutor retryExecutor = new RetryExecutor(
                        () -> this.executeByPkValues(runner, pkValues, option2Use), retryCount);
                result = retryExecutor.doWithRetry(result);
            }

            return result;

        } finally {
            // 7.释放锁
            if (runner.getCommand().getEnableLock()) {
                runner.unlock();
            }
        }
    }

    /**
     * 重试的执行器，目前仅对过期对象异常（违反乐观锁）才启动重试机制
     *
     * @author hangwen
     */
    private static class RetryExecutor {

        private int retryCount;
        private Supplier<CommandResult> retryAction;

        RetryExecutor(Supplier<CommandResult> retryAction, int retryCount) {
            this.retryAction = retryAction;
            this.retryCount = retryCount;
        }

        CommandResult doWithRetry(CommandResult result) {
            while (this.canRetry(result)) {
                result = retryAction.get();
            }

            return result;
        }

        private boolean canRetry(CommandResult result) {
            if (result.isSuccess()) {
                return false;
            }

            if (retryCount-- == 0) {
                return false;
            }

            return result.causeByStaleObject();
        }

    }

    private CommandResult executeByPkValues(CommandRunner<?> runner, List<Object> pkValues,
                                            CommandExecuteOption option) {
        // 1.按实体定义的属性，加载数据对象，这些对象是线程安全的
        Object[] dataObjects = this.loadDataObjects(runner, pkValues);

        // 2.如果option对象指定了dataObjectHandler对象，则会调用它的处理方法，可以对刚刚加载的数据进行处理
        ICommandExecuteDataObjectHandler dataObjectHandler = option.getDataObjectHandler();
        if (dataObjectHandler != null) {
            dataObjectHandler.accept(runner.getContext(), dataObjects);
        }

        // 3.执行核心流程
        return execute0(runner, dataObjects, option);
    }

    /**
     * 对实体执行命令，传入数据对象.
     * <p>
     * 如果command的{@link Command#isRequireReload()}设置为true,会再次重新从数据仓库加载数据，与传入的数据对象可能是不一致的。建议通过传入主键的方式进行调用。
     *
     * @param entity
     * @param command
     * @param dataObjects
     * @return
     */
    public CommandResult execute(Entity entity, Command command, Object[] dataObjects) {
        return execute(entity, command, dataObjects, DEFAULT_OPTION);
    }

    /**
     * 对实体执行命令，传入数据对象和命令执行选项.
     *
     * <p>
     * 如果command的{@link Command#isRequireReload()}设置为true,会再次重新从数据仓库加载数据，与传入的数据对象可能是不一致的。建议通过传入主键的方式进行调用。
     *
     * @param entity
     * @param command
     * @param dataObjects
     * @param option
     * @return
     */
    public CommandResult execute(Entity entity, Command command, Object[] dataObjects, CommandExecuteOption option) {
        assertParameters(entity, command);

        if (option == null) {
            option = DEFAULT_OPTION;
        }

        // 1.初始化命令运行器
        // 2.处理实体：对于有些实体，只需要部分属性
        CommandRunner<?> runner = CommandRunnerFactory.createRunner(entity, command, option);

        // 3.处理数据对象
        Object[] resolvedObjects = this.resolveDataObjects(runner, dataObjects);

        // 5.执行核心流程
        return execute0(runner, resolvedObjects, option);
    }

    /**
     * 对实体进行校验，校验器从command中获取，该方法可以用于获取合法的数据，剔除未通过校验的数据
     *
     * @param entity
     * @param command
     * @param dataObjects
     * @return
     */
    public ValidationErrors validate(Entity entity, Command command, Object[] dataObjects) {
        return this.validate(entity, command, dataObjects, DEFAULT_OPTION);
    }

    /**
     * 对实体进行校验，校验器从command中获取，该方法可以用于获取合法的数据，剔除未通过校验的数据
     *
     * @param entity
     * @param command
     * @param dataObjects
     * @return
     */
    public ValidationErrors validate(Entity entity, Command command, Object[] dataObjects,
                                     CommandExecuteOption option) {
        assertParameters(entity, command);

        if (option == null) {
            option = DEFAULT_OPTION;
        }

        // 1.初始化命令运行器
        // 2.处理实体：对于有些实体，只需要部分属性
        CommandRunner<?> runner = CommandRunnerFactory.createRunner(entity, command, option);

        // 3.处理数据对象
        Object[] resolvedObjects = this.resolveDataObjects(runner, dataObjects);

        try {
            // 执行校验，可能会抛出CommandExecuteRuntimeException
            return this.validate(runner, resolvedObjects);
        } finally {
            // 校验结束也要释放验证器资源
            this.afterInvoke(runner, resolvedObjects);
        }
    }

    private void afterInvoke(CommandRunner<?> runner, Object[] dataObjects) {
        try {
            runner.afterInvoke(dataObjects);
        } catch (Throwable e) {
            String message = this.genExceptionMessage(runner, "afterInvoke-001", e);
            logger.error(message, e);
        }
    }

    private void afterThrowing(CommandRunner<?> runner, Object[] dataObjects,
                               CommandExecuteRuntimeException exception) {
        try {
            runner.afterThrowing(dataObjects, exception);
        } catch (Throwable e) {
            String message = this.genExceptionMessage(runner, "afterThrowing-001", e);
            logger.error(message, e);
        }
    }

    private void assertParameters(Entity entity, Command command) {
        Assert.notNull(entity, "parameter 'entity' is required");
        Assert.notNull(command, "parameter 'command' is required");

        if (entity.getCommand(command.getName(), true) != command) {
            throw new IllegalArgumentException(
                    "command '" + command.toString() + "' dose not belong to '" + entity.toString() + "'");
        }
    }

    private void beforeInvoke(CommandRunner<?> runner, Object[] dataObjects) throws CommandExecuteRuntimeException {
        try {
            runner.beforeInvoke(dataObjects);
        } catch (CommandExecuteRuntimeException e) {
            throw e;
        } catch (Throwable e) {
            throw this.createException(runner, "beforeInvoke-001", e);
        }
    }

    private void checkStaleDataObjects(IDataField pkField, Object[] dataObjects, List<Object> pkValues, Entity entity) {
        if (dataObjects.length == pkValues.size()) {
            return;
        }

        Set<Object> pkValuesFromDB = Stream.of(dataObjects).map(o -> pkField.getValue(o)).collect(Collectors.toSet());
        List<Object> stalePkValues = pkValues.stream().filter(pk -> !pkValuesFromDB.contains(pk))
                .collect(Collectors.toList());

        String message = StringResources.get("pisces-domain.CommandExecutor.checkStaleDataObjects-001", stalePkValues,
                entity.toString());
        throw new StaleObjectException(message);
    }

    private void completeInvoke(CommandRunner<?> runner, Object[] dataObjects) {
        try {
            runner.completeInvoke(dataObjects);
        } catch (Throwable e) {
            String message = this.genExceptionMessage(runner, "completeInvoke-001", e);
            logger.error(message, e);
        }
    }

    private CommandExecuteRuntimeException createException(CommandRunner<?> runner, String messageKey,
                                                           Throwable cause) {
        String message = this.genExceptionMessage(runner, messageKey, cause);
        return new CommandExecuteRuntimeException(message, cause);
    }

    private CommandResult execute0(CommandRunner<?> runner, Object[] dataObjects, CommandExecuteOption option) {
        CommandResult result = null;

        try {
            result = execute1(runner, dataObjects, option);
            return result;
        } catch (CommandExecuteInvalidObjectException e) {
            // 事务内，验证未通过可能会抛出该异常，需要把它组装成result，这样finally块里依然可以发布事件
            result = CommandResult.error(e);
            throw e;
        } finally {
            // 发布领域事件
            if (result != null && !option.isDisableEvent()) {
                this.publishExecutedEvent(runner, dataObjects, result);
            }
        }
    }

    /**
     * 执行命令的核心流程
     *
     * @param runner
     * @param dataObjects
     * @param option
     * @return
     */
    private CommandResult execute1(CommandRunner<?> runner, Object[] dataObjects, CommandExecuteOption option) {
        try {
            // 执行前，命令和其插件可以对数据进行一些处理,可能会抛出CommandExecuteRuntimeException
            this.beforeInvoke(runner, dataObjects);

            // 执行校验，可能会抛出CommandExecuteRuntimeException
            ValidationErrors errors = this.validate(runner, dataObjects);
            if (errors != null && errors.hasError()) {
                // 没有通过校验
                return this.handleValidationErrors(runner, option, errors);
            }

            // 执行命令,可能会抛出CommandExecuteRuntimeException
            CommandResult commandResult = this.invoke(runner, dataObjects);

            // 执行完成，如有异常只会记录日志，不会抛出
            this.completeInvoke(runner, dataObjects);

            return commandResult;
        } catch (CommandExecuteRuntimeException e) {
            if (logger.isErrorEnabled()) {
                logger.error(e.getMessage(), e.getCause());
            }

            // 处理异常，如有执行过程异常只会记录日志，不会抛出
            this.afterThrowing(runner, dataObjects, e);

            if (TransactionHelper.inTransactionScope()) {
                // 处于事务内，及时抛出异常
                throw e;
            } else {
                // 友好返回结果
                return CommandResult.error(e);
            }
        } finally {
            // 执行结束，如有执行过程异常只会记录日志，不会抛出
            this.afterInvoke(runner, dataObjects);
        }
    }

    private String genExceptionMessage(CommandRunner<?> runner, String messageKey, Throwable cause) {
        CommandExecuteContext context = runner.getContext();

        String key = "pisces-domain.CommandExecutor." + messageKey;
        Entity entity = context.getEntity();
        Command command = context.getCommand();

        String message = StringResources.get(key, entity.toString(), command.toString(), cause.getMessage());
        return message;
    }

    private CommandResult handleValidationErrors(CommandRunner<?> runner, CommandExecuteOption option,
                                                 ValidationErrors errors) {
        CommandExecuteInvalidObjectException cause = runner.getCommand().wrapperValidationErrors(errors);

        if (cause == null) {
            cause = new CommandExecuteInvalidObjectException(errors.getErrors());
        }

        if (TransactionHelper.inTransactionScope() && option.isThrowExceptionWhileValidationFailInTransationScope()) {
            // 处于事务内，及时抛出异常
            throw cause;
        } else {
            // 友好返回结果
            return CommandResult.error(cause);
        }
    }

    private CommandResult invoke(CommandRunner<?> runner, Object[] dataObjects) throws CommandExecuteRuntimeException {
        try {
            // 执行具体的业务逻辑,启动事务
            return runner.invoke(dataObjects);
        } catch (CommandExecuteRuntimeException e) {
            throw e;
        } catch (Throwable e) {
            throw this.createException(runner, "invoke-001", e);
        }
    }

    private Object[] loadDataObjects(CommandRunner<?> runner, List<Object> pkValues) {
        Assert.notEmpty(pkValues, "parameter 'pkValues' is required");
        for (Object pkValue : pkValues) {
            Assert.notNull(pkValue, "parameter 'pkValues' has null value");
        }

        Entity entity = runner.getContext().getEntity();
        IDataObjectMeta objectMeta = entity.getDataObjectMeta();

        Object[] dataObjects = this.repository.loadBatchForUpdate(objectMeta, pkValues.toArray());
        this.checkStaleDataObjects(objectMeta.getPkField(), dataObjects, pkValues, entity);

        return dataObjects;
    }

    /**
     * 发布命令已执行事件，确保事件在事务完成后触发
     *
     * @param runner
     * @param dataObjects
     * @param result
     */
    private void publishExecutedEvent(CommandRunner<?> runner, Object[] dataObjects, CommandResult result) {
        Entity entity = runner.getContext().getEntity().fullEntity();
        CommandExecutedDomainEvent event = new CommandExecutedDomainEvent(entity, runner.getContext(), dataObjects,
                result);

        if (TransactionHelper.inTransactionScope()) {
            TransactionHelper.doAfterCompletion((status) -> {
                this.applicationEventPublisher.publishEvent(event);
            });
        } else {
            this.applicationEventPublisher.publishEvent(event);
        }
    }

    private Object[] resolveDataObjects(CommandRunner<?> runner, Object[] dataObjects) {
        Assert.notEmpty(dataObjects, "parameter 'dataObjects' is required");

        CommandExecuteContext context = runner.getContext();
        Command command = context.getCommand();

        Entity entity = context.getEntity();
        // 这里需要确保初始化objectMeta
        IDataObjectMeta objectMeta = entity.getDataObjectMeta();

        if (!command.getRequireReload()) {
            return dataObjects;
        }

        IDataField pkField = objectMeta.getPkField();

        List<Object> pkValues = Stream.of(dataObjects).map(o -> pkField.getValue(o)).collect(Collectors.toList());
        return this.loadDataObjects(runner, pkValues);
    }

    /**
     * 尝试获取锁，失败返回一个错误的CommandResult
     *
     * @param runner
     * @param pkValues
     * @param option
     * @return
     */
    private CommandResult tryLock(CommandRunner<?> runner, List<Object> pkValues, CommandExecuteOption option) {
        ValidationErrors errors = runner.lock(pkValues.toArray());
        if (errors == null || !errors.hasError()) {
            return null;
        }

        // 没有通过锁校验
        return this.handleValidationErrors(runner, option, errors);
    }

    private ValidationErrors validate(CommandRunner<?> runner, Object[] dataObjects)
            throws CommandExecuteRuntimeException {
        try {
            return runner.validate(dataObjects);
        } catch (CommandExecuteRuntimeException e) {
            throw e;
        } catch (Throwable e) {
            throw this.createException(runner, "validate-001", e);
        }
    }

}
